package com.atguigu.flinksql;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

public class Flink02_Table_BaseUse_Agg {
    public static void main(String[] args) {
         Configuration configuration = new Configuration();
         configuration.setInteger("rest.port",10000);
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
         env.setParallelism(2);

        DataStreamSource<WaterSensor> waterSensorStream =
                env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 2000L, 20),
                        new WaterSensor("sensor_2", 3000L, 30),
                        new WaterSensor("sensor_1", 4000L, 40),
                        new WaterSensor("sensor_1", 5000L, 50),
                        new WaterSensor("sensor_2", 6000L, 60));

        //1、创建表环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //2、将数据流转换为表
        Table table = tEnv.fromDataStream(waterSensorStream);

        //3、对表进行操作
        //在动态表上执行连续查询，会得到一个新的动态表
        Table select = table
                .groupBy($("id"))
                .select($("id"), $("vc").sum().as("vc_sum"));


        //4、将动态表重新转换为流进行操作
        //转换为retractStream之后会记录所有的聚合操作
        DataStream<Tuple2<Boolean, Row>> resultStream = tEnv.toRetractStream(select, Row.class);

        //对流中的数据进行处理

        resultStream.filter((FilterFunction<Tuple2<Boolean, Row>>) value -> value.f0).print();


        try {
             env.execute();
         } catch (Exception e) {
             e.printStackTrace();
         }
    }
}
